#include "RtBase.h"
#include "cmimconn.h"
#include "RtUtilTemplates.h"
#include "RtTrace.h"
#include "RtConnectionInterface.h"
#define MAX_RAW_CONN_WAIT_TIME	30000

///
IConnProvider* IConnProvider::Instance()
{ return CRtIMConnProvider::Instance(); }

///
CRtIMConnProvider* CRtIMConnProvider::Instance()
{ return CRtSingletonT<CRtIMConnProvider>::Instance(); }

CRtIMConnProvider::CRtIMConnProvider()
{
	m_cnfSink = NULL;
	m_p2pSink = NULL;
	m_pngSink = NULL;
	m_imSink = NULL;
	m_basicSink = NULL;
	Initialize();
}

CRtIMConnProvider::~CRtIMConnProvider()
{
	RT_INFO_TRACE("CRtIMConnProvider::~CRtIMConnProvider");
	UnInitialize();
}

void CRtIMConnProvider::GenerateMonitorInfo(CRtString& MonitorInfo)
{ CStatisticsData::GenerateMonitorInfo(MonitorInfo); }

RtResult CRtIMConnProvider::Initialize()
{
	RT_INFO_TRACE("CRtIMConnProvider::Initialize, ");

	CRtTimeValue tv(5.0);
	m_timer.Schedule(this, tv);

#ifdef RT_SUPPORT_BANDWIDTH_SERVER
	m_monitor = new CMonitorT<CRtIMConnProvider>(this);
	RtResult ret = m_monitor->Initialize("connmgr");
	if (RT_FAILED(ret))
	{
		RT_WARNING_TRACE("CRtIMConnProvider::Initialize, CMonitorT Initialize Failed. ret="<<ret);
	}
#endif
	return RT_OK;
}

RtResult CRtIMConnProvider::UnInitialize()
{
	RT_INFO_TRACE("CRtIMConnProvider::UnInitialize, m_connectors="<<m_connectors.size());

	m_timer.Cancel();

	IMConnectorsType::iterator it_cnr = m_connectors.begin();
	for (; it_cnr != m_connectors.end(); ++it_cnr)
	{
		CRtIMConnector* connector = *it_cnr;
		connector->ReleaseReference();
	}
	m_connectors.clear();

	StopListen();
	return RT_OK;
}

RtResult CRtIMConnProvider::StopListen()
{
	RT_INFO_TRACE("CRtIMConnProvider::StopListen, m_connections="<<m_connections.size());
	IMConnectionsType::iterator it_conn = m_connections.begin();
	for (; it_conn != m_connections.end(); ++it_conn)
	{
		CRtIMConnection* connection = *it_conn;
		connection->Disconnect(RT_OK);
		connection->ReleaseReference();
	}
	m_connections.clear();
	
	if(m_tcpAcceptBasic)
	{
		RT_INFO_TRACE("CRtIMConnProvider::Destroy, stop tcp basic");
		m_tcpAcceptBasic->StopListen(0);
		m_tcpAcceptBasic = NULL;
	}
	if(m_tcpAccept)
	{
		RT_INFO_TRACE("CRtIMConnProvider::Destroy, stop tcp");
		m_tcpAccept->StopListen(0);
		m_tcpAccept = NULL;
	}
	if(m_tcpAccept443)
	{
		RT_INFO_TRACE("CRtIMConnProvider::Destroy, stop tcp443");
		m_tcpAccept443->StopListen(0);
		m_tcpAccept443 = NULL;
	}
	if(m_tcpAccept80)
	{
		RT_INFO_TRACE("CRtIMConnProvider::Destroy, stop tcp80");
		m_tcpAccept80->StopListen(0);
		m_tcpAccept80 = NULL;
	}

	if(m_udpAcceptData)
	{
		RT_INFO_TRACE("CRtIMConnProvider::Destroy, stop data udp");
		m_udpAcceptData->StopListen(0);
		m_udpAcceptData = NULL;
	}
	if(m_tcpAcceptData)
	{
		RT_INFO_TRACE("CRtIMConnProvider::Destroy, stop data tcp");
		m_tcpAcceptData->StopListen(0);
		m_tcpAcceptData = NULL;
	}
	if(m_rudpAcceptData)
	{
		RT_INFO_TRACE("CRtIMConnProvider::Destroy, stop data rudp");
		m_rudpAcceptData->StopListen(0);
		m_rudpAcceptData = NULL;
	}

	return RT_OK;
}

RtResult CRtIMConnProvider::Listen(const TransportAddress& aAddress)
{
	CRtInetAddr addr, *pAddr = NULL;
	if(!aAddress.strIP.empty())
	{
		addr.Set(aAddress.strIP.c_str(),aAddress.nPort);
		pAddr = &addr;
	}
	else if (!IsLocalAddressSetted())
	{
		RT_WARNING_TRACE("CRtIMConnProvider::Listen, local address is empty!");
		return RT_ERROR_WRONG_STATUS;
	}
	return StartListenTcp(pAddr);
}

RtResult CRtIMConnProvider::ListenEx(const TransportAddress& aAddress, Event_Type nType)
{
	CRtInetAddr addr, *pAddr = NULL;
	if(!aAddress.strIP.empty())
	{
		addr.Set(aAddress.strIP.c_str(),aAddress.nPort);
		pAddr = &addr;
	}
	else if (!IsLocalAddressSetted())
	{
		RT_WARNING_TRACE("CRtIMConnProvider::Listen, local address is empty!");
		return RT_ERROR_WRONG_STATUS;
	}
	return StartListenTcpEx(pAddr, nType);
}

RtResult CRtIMConnProvider::AddedListen(const TransportAddress& aAddress)
{
	CRtInetAddr		  addr, *pAddr = NULL;
	Net_Protocol_Type type = NET_PROTOCOL_NONE;
	if(!aAddress.strIP.empty())
	{
		addr.Set(aAddress.strIP.c_str(),aAddress.nPort);
		pAddr = &addr;
		type = aAddress.nType;
	}
	else if(IsLocalAddressSetted())
	{
		type = NET_PROTOCOL_UDP;		
	}
	else if (!IsLocalAddressSetted())
	{
		RT_WARNING_TRACE("CRtIMConnProvider::AddedListen, local address is empty!");
		return RT_ERROR_WRONG_STATUS;
	}
	return StartListenData(pAddr, type);
}

RtResult CRtIMConnProvider::AddedListenEx(const TransportAddress& aAddress, Event_Type nType)
{
	CRtInetAddr		  addr, *pAddr = NULL;
	Net_Protocol_Type type = NET_PROTOCOL_NONE;
	if(!aAddress.strIP.empty())
	{
		addr.Set(aAddress.strIP.c_str(),aAddress.nPort);
		pAddr = &addr;
		type = aAddress.nType;
	}
	else if(IsLocalAddressSetted())
	{
		type = NET_PROTOCOL_TCP;		
	}
	else if (!IsLocalAddressSetted())
	{
		RT_WARNING_TRACE("CRtIMConnProvider::AddedListen, local address is empty!");
		return RT_ERROR_WRONG_STATUS;
	}
	return StartListenDataEx(pAddr, type, nType);
}

RtResult CRtIMConnProvider::StartListenTcp(CRtInetAddr* aAddr)
{
	if(m_tcpAccept)
	{ return RT_OK; }
	
	IRtAcceptor*				pAccept = NULL;
	CRtConnectionManager::CType nType = CRtConnectionManager::CTYPE_PDU_ANY
									  | CRtConnectionManager::CTYPE_PDU_KEEPALIVE
									  | CRtConnectionManager::CTYPE_TCP;
	RtResult					ret = RT_OK;
	ret = CRtConnectionManager::Instance()->CreateConnectionServer(
		nType,pAccept);
		RT_ASSERTE(RT_SUCCEEDED(ret));
		m_tcpAccept = pAccept;

	if (NULL == aAddr)
	{
		ret = CRtConnectionManager::Instance()->CreateConnectionServer(
			nType, pAccept);
		RT_ASSERTE(RT_SUCCEEDED(ret));
		m_tcpAccept443 = pAccept;
		
		ret = CRtConnectionManager::Instance()->CreateConnectionServer(
			nType, pAccept);
		RT_ASSERTE(RT_SUCCEEDED(ret));
		m_tcpAccept80 = pAccept;

		m_tcpAddress.Set(m_localAddress.c_str(), 443);
		if(RT_FAILED((ret = m_tcpAccept443->StartListen(this, m_tcpAddress))))
			m_tcpAccept443 = NULL;
		
		m_tcpAddress.Set(m_localAddress.c_str(), 80);
		if(RT_FAILED((ret = m_tcpAccept80->StartListen(this, m_tcpAddress))))
			m_tcpAccept80 = NULL;

		for(WORD i = 1024; i < 1500; ++i)
		{
		m_tcpAddress.Set(m_localAddress.c_str(), i);
		if(RT_SUCCEEDED((ret = m_tcpAccept->StartListen(this, m_tcpAddress))))
			{ return RT_OK; }
		}
		m_tcpAccept = NULL;
	}
	else
	{
		m_tcpAddress = *aAddr;
		if(RT_SUCCEEDED((ret = m_tcpAccept->StartListen(this, m_tcpAddress))))
		{ return RT_OK; }

		m_tcpAccept = NULL;
		if(IsLocalAddressSetted())
		{ return StartListenTcp(NULL); }
	}
	return ret;
}

RtResult CRtIMConnProvider::StartListenTcpEx(CRtInetAddr* aAddr, Event_Type eVentType)
{
	if(m_tcpAccept)
	{ return RT_OK; }
	
	IRtAcceptor*				pAccept = NULL;
	CRtConnectionManager::CType nType;
	if (EVENT_TYPE_BASIC_TCP == eVentType)
	{
		nType = CRtConnectionManager::CTYPE_TCP;
	}
	else{
		nType = CRtConnectionManager::CTYPE_PDU_ANY
				| CRtConnectionManager::CTYPE_PDU_KEEPALIVE
				| CRtConnectionManager::CTYPE_TCP;
	}
	RtResult ret = RT_OK;
	ret = CRtConnectionManager::Instance()->CreateConnectionServer(nType, pAccept);
	RT_ASSERTE(RT_SUCCEEDED(ret));
	m_tcpAccept = pAccept;
	
	if (NULL == aAddr)
	{
		ret = CRtConnectionManager::Instance()->CreateConnectionServer(
			nType, pAccept);
		RT_ASSERTE(RT_SUCCEEDED(ret));
		m_tcpAccept443 = pAccept;
		
		ret = CRtConnectionManager::Instance()->CreateConnectionServer(
			nType, pAccept);
		RT_ASSERTE(RT_SUCCEEDED(ret));
		m_tcpAccept80 = pAccept;
		
		m_tcpAddress.Set(m_localAddress.c_str(), 443);
		if(RT_FAILED((ret = m_tcpAccept443->StartListen(this, m_tcpAddress))))
			m_tcpAccept443 = NULL;
		
		m_tcpAddress.Set(m_localAddress.c_str(), 80);
		if(RT_FAILED((ret = m_tcpAccept80->StartListen(this, m_tcpAddress))))
			m_tcpAccept80 = NULL;
		
		for(WORD i = 1024; i < 1500; ++i)
		{
			m_tcpAddress.Set(m_localAddress.c_str(), i);
			if(RT_SUCCEEDED((ret = m_tcpAccept->StartListen(this, m_tcpAddress))))
			{ return RT_OK; }
		}
		m_tcpAccept = NULL;
	}
	else
	{
		m_tcpAddress = *aAddr;
		if(RT_SUCCEEDED((ret = m_tcpAccept->StartListen(this, m_tcpAddress))))
		{ return RT_OK; }
		
		m_tcpAccept = NULL;
		if(IsLocalAddressSetted())
		{ return StartListenTcpEx(NULL, EVENT_TYPE_BASIC_TCP); }
	}
	return ret;
}

RtResult CRtIMConnProvider::StartListenData(CRtInetAddr* aAddr, Net_Protocol_Type aType)
{
	IRtAcceptor*				pAccept = NULL;
	CRtAutoPtr<IRtAcceptor>*	ppAcceptor = NULL;
	CRtConnectionManager::CType nType;
	switch(aType)
	{
	case NET_PROTOCOL_UDP:
		{
			if(!aAddr && m_udpAcceptData.Get())	// default udp only listen once
				return RT_OK;
			
			pAccept = m_udpAcceptData.Get();
			ppAcceptor = &m_udpAcceptData;
			nType = CRtConnectionManager::CTYPE_UDP|CRtConnectionManager::CTYPE_PDU_PACKAGE|CRtConnectionManager::CTYPE_PDU_KEEPALIVE;
		}
		break;
	case NET_PROTOCOL_RUDP:
		{		
			if(!aAddr)
				return RT_ERROR_INVALID_ARG;
			pAccept = m_rudpAcceptData.Get();
			ppAcceptor = &m_rudpAcceptData;
			nType = CRtConnectionManager::CTYPE_RUDP;
		}
		break;
	case NET_PROTOCOL_TCP:
	default:
		{		
			if(!aAddr)
				return RT_ERROR_INVALID_ARG;
			pAccept = m_tcpAcceptData.Get();
			ppAcceptor = &m_tcpAcceptData;
			nType = CRtConnectionManager::CTYPE_TCP|CRtConnectionManager::CTYPE_PDU_PACKAGE|CRtConnectionManager::CTYPE_PDU_KEEPALIVE;
		}
		break;
	}
	
	RtResult ret = RT_OK;	
	if (!pAccept)
	{
		ret = CRtConnectionManager::Instance()->CreateConnectionServer(nType, pAccept);
		if(RT_FAILED(ret))
			return ret;
	}
	RT_ASSERTE(pAccept);	
	
	if(!aAddr)	// for default udp listen
	{
		for(WORD i = 5257; i < 50000; i++)
		{
			m_udpAddress.Set(m_localAddress.c_str(), i);
			if(RT_SUCCEEDED((ret = pAccept->StartListen(this, m_udpAddress))))
			{ return RT_OK; }
		}
	}
	else		// for rudp or tcp listen
	{
		m_udpAddress = *aAddr;
		if(RT_SUCCEEDED((ret = pAccept->StartListen(this, m_udpAddress))))
		{ return RT_OK; }
	}
	
	(* ppAcceptor) = pAccept;
	return ret;
}

RtResult CRtIMConnProvider::StartListenDataEx(CRtInetAddr* aAddr, Net_Protocol_Type aType, Event_Type eVentType)
{
	IRtAcceptor*				pAccept = NULL;
	CRtAutoPtr<IRtAcceptor>*	ppAcceptor = NULL;
	CRtConnectionManager::CType nType;
	switch(aType)
	{
	case NET_PROTOCOL_UDP:
		{
			if(!aAddr && m_udpAcceptData.Get())	// default udp only listen once
				return RT_OK;
			
			pAccept = m_udpAcceptData.Get();
			ppAcceptor = &m_udpAcceptData;
			nType = CRtConnectionManager::CTYPE_UDP|CRtConnectionManager::CTYPE_PDU_PACKAGE|CRtConnectionManager::CTYPE_PDU_KEEPALIVE;
		}
		break;
	case NET_PROTOCOL_RUDP:
		{		
			if(!aAddr)
				return RT_ERROR_INVALID_ARG;
			pAccept = m_rudpAcceptData.Get();
			ppAcceptor = &m_rudpAcceptData;
			nType = CRtConnectionManager::CTYPE_RUDP;
		}
		break;
	case NET_PROTOCOL_TCP:
	default:
		{
			if(!aAddr)
				return RT_ERROR_INVALID_ARG;
			pAccept = m_tcpAcceptData.Get();
			ppAcceptor = &m_tcpAcceptData;
			if (EVENT_TYPE_BASIC_TCP == eVentType)
			{
				nType = CRtConnectionManager::CTYPE_TCP;
			}
			else
				nType = CRtConnectionManager::CTYPE_TCP|CRtConnectionManager::CTYPE_PDU_PACKAGE|CRtConnectionManager::CTYPE_PDU_KEEPALIVE;
		}
		break;
	}
	
	RtResult ret = RT_OK;	
	if (!pAccept)
	{
		ret = CRtConnectionManager::Instance()->CreateConnectionServer(nType, pAccept);
		if(RT_FAILED(ret))
			return ret;
	}
	RT_ASSERTE(pAccept);	
	
	if(!aAddr)	// for default udp listen
	{
		for(WORD i = 5257; i < 50000; i++)
		{
			m_udpAddress.Set(m_localAddress.c_str(), i);
			if(RT_SUCCEEDED((ret = pAccept->StartListen(this, m_udpAddress))))
			{ return RT_OK; }
		}
	}
	else		// for rudp or tcp listen
	{
		m_udpAddress = *aAddr;
		if(RT_SUCCEEDED((ret = pAccept->StartListen(this, m_udpAddress))))
		{ return RT_OK; }
	}
	
	(* ppAcceptor) = pAccept;
	return ret;
}

RtResult CRtIMConnProvider::Register(Event_Type nType,
								IConnProviderSink* pSink)
{
	switch(nType)
	{
	case EVENT_TYPE_P2P_PING:
		m_pngSink = pSink;
		break;
	case EVENT_TYPE_CONFERENCE_PING:
		m_cnfSink = pSink;
		break;
	case EVENT_TYPE_P2P_SHAREDATA:
		m_p2pSink = pSink;
		break;
	case EVENT_TYPE_IM:
		m_imSink = pSink;
		break;
	case EVENT_TYPE_BASIC_TCP:
		m_basicSink = pSink;
		break;
	default:
		return RT_ERROR_NOT_IMPLEMENTED;
	}
	return RT_OK;
}

RtResult CRtIMConnProvider::GetListenAddress(
		BOOL bPrimary,
		TransportAddress& address)
{
	if(bPrimary)
	{
		address.strIP = m_tcpAddress.GetIpDisplayName();
		address.nPort = m_tcpAddress.GetPort();
		address.nType = NET_PROTOCOL_TCP;
	}
	else
	{
		address.strIP = m_udpAddress.GetIpDisplayName();
		address.nPort = m_udpAddress.GetPort();
		address.nType = NET_PROTOCOL_UDP;
	}
	return RT_OK;
}

RtResult CRtIMConnProvider::Connect(
		IConnProviderSink* pSink,
		DWORD aNumber,
		const TransportAddress* aAddresses,
		Event_Type aEventType,
		CRtTimeValue *aTimeout,
		BOOL aSync)
{
	RT_INFO_TRACE("CRtIMConnProvider::Connect, pSink="<<pSink
		<<",address aNumber="<<aNumber<<",aEventType="<<aEventType<<",aSync="<<aSync);
	return Connect_i(pSink, aNumber, aAddresses, aEventType, aTimeout, aSync, TRUE);
}
RtResult CRtIMConnProvider::ConnectData(
		IConnProviderSink* pSink, 
		DWORD aNumber,
		TransportAddress* aAddresses,
		Event_Type aEventType)
{
	RT_INFO_TRACE("CRtIMConnProvider::ConnectData, pSink="<<pSink
		<<",address aNumber="<<aNumber<<",aEventType="<<aEventType);
	CRtTimeValue aTimeout((5+aNumber),0);
	return Connect_i(pSink, aNumber, aAddresses, aEventType, &aTimeout, FALSE, FALSE);
}

RtResult CRtIMConnProvider::Connect_i(
		IConnProviderSink* pSink,
		DWORD  aNumber,
		const TransportAddress* aAddresses,
		Event_Type aEventType,
		CRtTimeValue *aTimeout,
		BOOL aSync,
		BOOL aCommand)
{
	RT_ASSERTE_RETURN(0 != aNumber && aAddresses && pSink, RT_ERROR_INVALID_ARG);

	IRtDetectionConnector*	pConnector = NULL;
	RtResult ret = CRtConnectionManager::Instance()->CreateDetectionConnectionClient(pConnector);
	if(RT_FAILED(ret))
		return ret;

	CRtConnectionManager::CType nType;
	CRtInetAddr addr;
	for(DWORD i = 0; i < aNumber; i++)
	{
		switch(aAddresses[i].nType)
		{
		case NET_PROTOCOL_UDP:
			nType = CRtConnectionManager::CTYPE_PDU_PACKAGE|CRtConnectionManager::CTYPE_PDU_KEEPALIVE|CRtConnectionManager::CTYPE_UDP;
			break;
		case NET_PROTOCOL_TCP:
			{
				if (aCommand)
				{
					if (EVENT_TYPE_IM == aEventType)
					{
						RT_INFO_TRACE_THIS("CRtIMConnProvider::Connect_i, use reliable tcp.");
// 						nType = CRtConnectionManager::CTYPE_TCP|CRtConnectionManager::CTYPE_PDU_LENGTH;
						nType = CRtConnectionManager::CTYPE_TCP|CRtConnectionManager::CTYPE_PDU_RELIABLE;						
					}
					else if (EVENT_TYPE_P2P_SHAREDATA == aEventType)
						nType = CRtConnectionManager::CTYPE_TCP|CRtConnectionManager::CTYPE_PDU_PACKAGE | CRtConnectionManager::CTYPE_PDU_KEEPALIVE;
					else if (EVENT_TYPE_BASIC_TCP == aEventType)
					{
						RT_INFO_TRACE_THIS("CRtIMConnProvider::Connect_i, use basic tcp.");
						nType = CRtConnectionManager::CTYPE_TCP;
					}
					else // for test
					{
						RT_INFO_TRACE_THIS("CRtIMConnProvider::Connect_i, use reliable tcp.");
// 						nType = CRtConnectionManager::CTYPE_TCP|CRtConnectionManager::CTYPE_PDU_PACKAGE|CRtConnectionManager::CTYPE_PDU_KEEPALIVE;
						nType = CRtConnectionManager::CTYPE_TCP|CRtConnectionManager::CTYPE_PDU_RELIABLE;
					}
				}
				else
				{
	 				nType = CRtConnectionManager::CTYPE_PDU_PACKAGE|CRtConnectionManager::CTYPE_PDU_KEEPALIVE|CRtConnectionManager::CTYPE_TCP;
				}
				addr.Set(aAddresses[i].strIP.c_str(),aAddresses[i].nPort);
				CRtTimeValue tmptime(3,0);
				pConnector->AddConnection(nType, addr, &tmptime);

				if (aCommand)
				{
					if (EVENT_TYPE_IM == aEventType)
						nType = CRtConnectionManager::CTYPE_TCP_WITH_BROWER_PROXY|CRtConnectionManager::CTYPE_PDU_RELIABLE;
					else if (EVENT_TYPE_P2P_SHAREDATA == aEventType)
						nType = CRtConnectionManager::CTYPE_TCP_WITH_BROWER_PROXY|CRtConnectionManager::CTYPE_PDU_PACKAGE | CRtConnectionManager::CTYPE_PDU_KEEPALIVE;
					else if (EVENT_TYPE_BASIC_TCP == aEventType)
						nType = CRtConnectionManager::CTYPE_TCP_WITH_BROWER_PROXY|CRtConnectionManager::CTYPE_TCP;
					else
// 						nType = CRtConnectionManager::CTYPE_TCP_WITH_BROWER_PROXY|CRtConnectionManager::CTYPE_PDU_PACKAGE|CRtConnectionManager::CTYPE_PDU_KEEPALIVE;
						nType = CRtConnectionManager::CTYPE_TCP_WITH_BROWER_PROXY|CRtConnectionManager::CTYPE_PDU_RELIABLE;
				}
				else
				{
					nType = CRtConnectionManager::CTYPE_PDU_PACKAGE|CRtConnectionManager::CTYPE_PDU_KEEPALIVE|CRtConnectionManager::CTYPE_TCP_WITH_BROWER_PROXY;
				}
			}
			break;
		case NET_PROTOCOL_RUDP:
			nType = CRtConnectionManager::CTYPE_RUDP;
			break;
		default:
			RT_INFO_TRACE("CRtIMConnProvider::Connect, WRONG CONNECTION type");
			break;
		}
		addr.Set(aAddresses[i].strIP.c_str(),aAddresses[i].nPort);
		CRtTimeValue tmptime(3,0);
		pConnector->AddConnection(nType, addr, &tmptime);
	}
	CRtIMConnector* connector = new CRtIMConnector(
		*this, pConnector, pSink, aEventType);
	connector->SetAddress(addr);
	connector->AddReference();
	m_connectors.push_back(connector);
	IRtDetectionConnector::DetectMode inMode = aSync ? IRtDetectionConnector::DETECT_CONCURRENT : IRtDetectionConnector::DETECT_INTERVAL_1S;
	pConnector->StartDetectionConnect(connector, aTimeout, inMode);
	return RT_OK;
}
// 
// RtResult CRtIMConnProvider::ConnectRudpAndTcp(IConnProviderSink* pSink,
// 											  const CRtInetAddr& addr,
// 											  Event_Type nEventType,
// 											  CRtTimeValue *aTimeout)
// {
// 	RT_INFO_TRACE("CRtIMConnProvider::ConnectRudpAndTcp, ");
// 	IRtDetectionConnector*	pConnector = NULL;
// 	RtResult ret = CRtConnectionManager::Instance()->CreateDetectionConnectionClient(pConnector);
// 	if(RT_FAILED(ret))
// 		return ret;
// 
// 	CRtConnectionManager::CType nType = CRtConnectionManager::CTYPE_RUDP;
// 	pConnector->AddConnection(nType, addr, aTimeout);
// 
// 	nType = CRtConnectionManager::CTYPE_PDU_LENGTH | CRtConnectionManager::CTYPE_TCP_WITH_BROWER_PROXY;
// 	pConnector->AddConnection(nType, addr, aTimeout);
// 
// 	CRtIMConnector* pMyConnector = new CRtIMConnector(this,pConnector, pSink, nEventType);
// 	pMyConnector->SetAddress(addr);
// 	pMyConnector->AddReference();
// 	m_connectors.push_back(pMyConnector);
// 	pConnector->StartDetectionConnect(pMyConnector, aTimeout, IRtDetectionConnector::DETECT_INTERVAL_1S);
// 
// 	return RT_OK;
// }

RtResult CRtIMConnProvider::CancelConnect(
		IConnProviderSink* pSink, 
		TransportAddress* aAddress,
		BOOL bHaveMultiPort)
{
	RT_INFO_TRACE("CRtIMConnProvider::CancelConnect, pSink="<<pSink
		<<",address="<<(aAddress ? aAddress->strIP : "")<<",multiport="<<bHaveMultiPort);
	IMConnectorsType::iterator it = m_connectors.begin();
	while (it != m_connectors.end())
	{
		BOOL			bHave = FALSE;
		CRtIMConnector* pConnector = (*it);
		if (pConnector->GetSink() == pSink)
		{
			if (aAddress && pConnector->GetAddress().GetIpDisplayName() == aAddress->strIP)
			{
				if (bHaveMultiPort
				|| (pConnector->GetAddress().GetPort() == aAddress->nPort))
				{ bHave = TRUE; }
			}
			else
			{	  bHave = TRUE; }
		}

		if (bHave)
		{
			pConnector->CancelConnect();
			pConnector->ReleaseReference();

			IMConnectorsType::iterator it_tmp = it;
			++ it;
			m_connectors.erase(it_tmp);
		}
		else
		{	++ it;	}
	}
	return RT_OK;
}

void CRtIMConnProvider::RemoveConnector(
		CRtIMConnector* aConnector,
		BOOL bDelete)
{
	IMConnectorsType::iterator it = m_connectors.begin();
	for (; it != m_connectors.end(); ++it)
	{
		CRtIMConnector* findConnector = *it;
		if(findConnector == aConnector)
		{
			m_connectors.erase(it);
			if(bDelete)
			{ findConnector->ReleaseReference(); }
			break;
		}
	}
}

void CRtIMConnProvider::OnConnectIndication(
											RtResult aReason,
											IRtTransport *aTrpt,
											IRtAcceptorConnectorId *aRequestId)
{
	RT_ASSERTE(RT_SUCCEEDED(aReason));
	
	CRtAutoPtr<CRtIMConnection> connection = new CRtIMConnection(aTrpt);
	DWORD						transType = 0;
	aTrpt->GetOption(RT_OPT_TRANSPORT_TRAN_TYPE, &transType);
	if ( !RT_BIT_ENABLED(transType, CRtConnectionManager::CTYPE_PDU_MASK) )
	{
		if (m_basicSink)
			m_basicSink->OnConnectionReady(aReason, connection.ParaIn(), NULL);
	}
	else if (transType & CRtConnectionManager::CTYPE_UDP)
	{
		if(m_cnfSink)
			m_cnfSink->OnConnectionReady(RT_OK, connection.ParaIn(), NULL);
	}
	else
	{
		connection->AddReference();
		connection->Initialize(this);
		m_connections.push_back(connection.Get());
	}
}

void CRtIMConnProvider::OnReceiveData(CRtMessageBlock& aData, IConnection* pConn)
{
	CRtString shandData = aData.FlattenChained();
	LPBYTE	  handData = (LPBYTE)shandData.c_str();
	if (4 != shandData.length()
		|| handData[0] != 0xFF || handData[1] != 0xFE || handData[2] != 0xFD)
	{
		RT_WARNING_TRACE("CRtIMConnProvider::OnReceiveData, receive error hand pdu. data: "<<shandData);
		return ;
	}
	else if (handData[3] == (char)EVENT_TYPE_P2P_PING)
	{
		if(m_pngSink)
			m_pngSink->OnConnectionReady(RT_OK, pConn, NULL);
	}
	else if(handData[3] == (char)EVENT_TYPE_CONFERENCE_PING)
	{
		if(m_cnfSink)
			m_cnfSink->OnConnectionReady(RT_OK, pConn, NULL);
	}
	else if(handData[3] == (char)EVENT_TYPE_P2P_SHAREDATA)
	{
		if(m_p2pSink)
			m_p2pSink->OnConnectionReady(RT_OK, pConn, NULL);
	}
	else
	{
		RT_ASSERTE(FALSE);
	}
	//remove from raw connection list;
	OnDisconnect(RT_OK, pConn);
}

void CRtIMConnProvider::OnDisconnect(RtResult aReason, IConnection* aConn)
{
	IMConnectionsType::iterator it = m_connections.begin();
	for(; it != m_connections.end(); ++it)
	{
		CRtIMConnection* connection = *it;
		if(connection == aConn)
		{
			m_connections.erase(it);
			connection->ReleaseReference();
			break;
		}
	}
}

void CRtIMConnProvider::OnTimer(CRtTimerWrapper* aID)
{
	LONGLONG					currentTime = GetTickCountEx();
	IMConnectionsType::iterator it_begin = m_connections.begin(), it;
	
	for(it=it_begin; m_connections.end() != it; ++it)
	{
		CRtIMConnection* connection = *it;
		if (currentTime > connection->GetStartTime() 
			&& (currentTime - connection->GetStartTime()) > MAX_RAW_CONN_WAIT_TIME)
		{
			connection->ReleaseReference();
		}
		else
		{
			if(it != it_begin)
			{ m_connections.erase(it_begin, it); }
			break;
		}
	}
}

/// class CRtIMConnector
CRtIMConnector::CRtIMConnector(
		CRtIMConnProvider& aProvider,
		IRtConnector* aConnector,
		IConnProviderSink* aSink,
		Event_Type aEventType)
	: m_provider(aProvider)
	, m_connector(aConnector)
	, m_pSink(aSink)
	, m_eventType(aEventType)
{
	RT_INFO_TRACE_THIS("CRtIMConnector::CRtIMConnector, aSink="<<aSink<<",aEventType="<<aEventType<<",aConnector="<<aConnector);
	RT_ASSERTE(aConnector);
}
CRtIMConnector::~CRtIMConnector()
{
	RT_INFO_TRACE_THIS("CRtIMConnector::~CRtIMConnector, m_pSink="<<m_pSink<<",m_connector="<<m_connector.Get());
	CancelConnect();
	m_pSink = NULL;
}

RtResult CRtIMConnector::Connect(const CRtInetAddr& aAddress, CRtTimeValue *aTimeout)
{
	m_address = aAddress;
	m_connector->AsycConnect(this, m_address, aTimeout);
	return RT_OK;
}

void CRtIMConnector::CancelConnect()
{
	if(m_connector)
	{ 
		m_connector->CancelConnect();
		m_connector = NULL;
	}	
}

void CRtIMConnector::OnConnectIndication(
		RtResult aReason,
		IRtTransport *aTrpt,
		IRtAcceptorConnectorId *aRequestId)
{
	CRtAutoPtr<CRtIMConnector>	self(this);
	CRtAutoPtr<CRtIMConnection> connection;
	CRtInetAddr					peerAddress;
	if(m_pSink && RT_SUCCEEDED(aReason))
	{
		aTrpt->GetOption(RT_OPT_TRANSPORT_PEER_ADDR, &peerAddress);
		connection = new CRtIMConnection(aTrpt);
		RT_ASSERTE(connection);

		DWORD transType = 0;
		aTrpt->GetOption(RT_OPT_TRANSPORT_TRAN_TYPE, &transType);
		if (!(transType & CRtConnectionManager::CTYPE_UDP) // not udp
		 && (m_eventType != EVENT_TYPE_IM) && (m_eventType != EVENT_TYPE_RUDP) 
		 && (RT_BIT_ENABLED(transType, CRtConnectionManager::CTYPE_PDU_MASK)))
		{
			BYTE data[4]={0xFF,0xFE,0xFD,(BYTE)m_eventType};
			CRtMessageBlock handex(4, (LPCSTR)data, CRtMessageBlock::DONT_DELETE, 4);
			aTrpt->SendData(handex);
		}

		if (!m_provider.IsLocalAddressSetted() 
		 && (transType & CRtConnectionManager::CTYPE_TCP))
		{
			CRtInetAddr localAddress;
			RtResult ret = aTrpt->GetOption(RT_OPT_TRANSPORT_LOCAL_ADDR, &localAddress);
			if (RT_FAILED(ret)) 
			{
				RT_WARNING_TRACE_THIS("CRtIMConnector::OnConnectIndication, localaddr="
					<<localAddress.GetIpDisplayName()<<", GetOption ret="<<ret);
			}
			else 
			{ 
				GetLocalAddress(localAddress);
				RT_INFO_TRACE_THIS("CRtIMConnector::OnConnectIndication, localaddr="
					<<localAddress.GetIpDisplayName());
				m_provider.SetLocalAddress(localAddress.GetIpDisplayName());
			}
		}
	}
	if (m_pSink)
		m_pSink->OnConnectionReady(aReason, connection.ParaIn(), &peerAddress);
	m_provider.RemoveConnector(this, TRUE);
}

void CRtIMConnector::GetLocalAddress(CRtInetAddr& localAddress)
{
#ifdef RT_WIN32
	if(localAddress.GetIpDisplayName() == "127.0.0.1")
	{
		char achBuf[256];
		if (gethostname(achBuf, sizeof(achBuf)) == 0)
		{
			HOSTENT* pHost = gethostbyname(achBuf);
			if (NULL != pHost)
			{
				int i = 0;
				while ( pHost->h_addr_list[i] )
				{
					struct in_addr netip1;
					memcpy(&netip1.s_addr, pHost->h_addr_list[i], pHost->h_length);
					char* p = inet_ntoa(netip1);
					if(p && strcmp(p, "127.0.0.1") == 0)
						i++;
					else
					{
						localAddress.SetIpAddrByString(p);
						break;
					}
				}
			}
		}
	}
#endif // RT_WIN32
}
//Statistics
DWORD			CStatisticsData::g_allSendData  = 0;
DWORD			CStatisticsData::g_allRecvData  = 0;
DWORD			CStatisticsData::g_lastSendData = 0;
DWORD			CStatisticsData::g_lastRecvData = 0;
LONGLONG		CStatisticsData::g_lastGetTick = GetTickCountEx();
std::list<CStatisticsData* > CStatisticsData::g_allStatistics;

void CStatisticsData::GenerateMonitorInfo(CRtString& monitorInfo)
{
	// all connection statistics info
	char	 szbuf[256] = {0};
	DWORD timespan = DWORD(GetTickCountEx() - g_lastGetTick);
	snprintf(szbuf, sizeof(szbuf), "CONN MGT[all connection info]: time=%d  cur r:%d s:%d; all r:%d s:%d \n", 
		timespan, 
		g_lastRecvData, 
		g_lastSendData, 
		g_allRecvData,
		g_allSendData);
	g_lastGetTick = GetTickCountEx();
	g_lastSendData = 0;
	g_lastRecvData = 0;
	monitorInfo += szbuf;

	// every connection statistics info
	std::list<CStatisticsData* >::iterator it = g_allStatistics.begin();
	for(; g_allStatistics.end() != it; ++it)
	{
		(*it)->GenerateOneMonitorInfo(monitorInfo);
	}
}

void CStatisticsData::GenerateOneMonitorInfo(CRtString& monitorInfo)
{
	RT_ASSERTE_RETURN_VOID(m_conn);

	char	 szbuf[256] = {0};
	snprintf(szbuf, sizeof(szbuf), "	[conn 0x%x loc:%s:%d peer:%s:%d]: cur r:%d s:%d; all r:%d s:%d \n", 
		m_conn,
		m_conn->GetLoaclAddr().strIP.c_str(),
		m_conn->GetLoaclAddr().nPort,
		m_conn->GetPeerAddr().strIP.c_str(),
		m_conn->GetPeerAddr().nPort,
		m_lastRecvData, 
		m_lastSendData,
		m_allRecvData,
		m_allSendData);
	m_lastRecvData = 0;
	m_lastSendData = 0;
	monitorInfo.append(szbuf);
}

CStatisticsData::CStatisticsData(CRtIMConnection* conn)
{
	RT_ASSERTE_RETURN_VOID(conn);
	m_conn = conn;
	m_allSendData = 0;
	m_allRecvData = 0;
	m_lastSendData = 0;
	m_lastRecvData = 0;

	g_allStatistics.push_back(this);
}

CStatisticsData::~CStatisticsData()
{
	g_allStatistics.remove(this);
}

void CStatisticsData::SendData(DWORD dataLen)
{
	m_allSendData += dataLen;
	m_lastSendData += dataLen;

	g_lastSendData += dataLen;
	g_allSendData += dataLen;
}

void CStatisticsData::ReceiveData(DWORD dataLen)
{
	m_allRecvData += dataLen;
	m_lastRecvData += dataLen;

	g_lastRecvData += dataLen;
	g_allRecvData += dataLen;
}

/// class CRtIMConnection
#ifdef RT_SUPPORT_BANDWIDTH
DWORD	CRtIMConnection::s_dwSendData = 0;
DWORD	CRtIMConnection::s_dwRecvData = 0;
LONGLONG	CRtIMConnection::s_dwLastSendGetTick = 0;
LONGLONG	CRtIMConnection::s_dwLastRecvGetTick = 0;
DWORD	CRtIMConnection::s_curSendBand = 0;
DWORD	CRtIMConnection::s_curRecvBand = 0;
CRtIMConnection::ConnDatasType CRtIMConnection::s_dwSendDatas;
CRtIMConnection::ConnDatasType CRtIMConnection::s_dwRecvDatas;
#endif

CRtIMConnection::CRtIMConnection(IRtTransport* aTransport)
#if defined(RT_SUPPORT_BANDWIDTH_SERVER) || defined(RT_SUPPORT_BANDWIDTH)
	: m_statistic(this)
#endif
{
// 	RT_INFO_TRACE_THIS_DETAIL("CRtIMConnection::CRtIMConnection trpt="<<aTransport);
	RT_ASSERTE(aTransport);
	m_transport = aTransport;
	m_transport->OpenWithSink(this);
	m_pSink = NULL;
	m_bCanSend = TRUE;
	m_startTime = GetTickCountEx();

	GetAddress(RT_OPT_TRANSPORT_LOCAL_ADDR, m_addrLocal);
	GetAddress(RT_OPT_TRANSPORT_PEER_ADDR, m_addrPeer);
}
CRtIMConnection::~CRtIMConnection()
{
 	RT_INFO_TRACE_THIS("CRtIMConnection::~CRtIMConnection m_bCanSend="
 		<<m_bCanSend<<",sink="<<m_pSink<<",trpt="<<m_transport.Get());
	if (m_transport)
	{
		m_transport->Disconnect(0);
		m_transport = NULL;
	}
	m_pSink = NULL;
}

void CRtIMConnection::OnReceive(
		CRtMessageBlock &aData,
		IRtTransport *aTrptId,
		CRtTransportParameter *aPara)
{
#ifdef RT_SUPPORT_BANDWIDTH
	if(s_dwLastRecvGetTick)
		s_dwRecvData += aData.GetChainedLength();
#endif

#if defined(RT_SUPPORT_BANDWIDTH_SERVER) || defined(RT_SUPPORT_BANDWIDTH)
	m_statistic.ReceiveData(aData.GetChainedLength());
#endif
	if (m_pSink)
	{ m_pSink->OnReceiveData(aData, this); }
}


void CRtIMConnection::OnSend(
		IRtTransport *aTrptId,
		CRtTransportParameter *aPara)
{
// 	RT_INFO_TRACE_THIS_DETAIL("CRtIMConnection::OnSend m_bCanSend="
// 		<<m_bCanSend<<",sink="<<m_pSink<<",trpt="<<m_transport.Get());
	if (!m_bCanSend)
	{ m_bCanSend = TRUE; }
	if (m_pSink)
	{ m_pSink->OnSend(this); }
}

void CRtIMConnection::OnDisconnect(
		RtResult aReason,
		IRtTransport *aTrptId)
{
	RT_INFO_TRACE_THIS("CRtIMConnection::OnDisconnect trpt="<<aTrptId<<", aReason="<<aReason);
	if (m_transport)
	{ 
		m_transport->Disconnect(0); 
		m_transport = NULL;
	}
	m_bCanSend = FALSE;
	if (m_pSink)
	{ m_pSink->OnDisconnect(aReason, this); }
}
	
RtResult CRtIMConnection::Initialize(IConnectionSink* pSink)
{
	m_pSink = pSink;
	return RT_OK;
}

RtResult CRtIMConnection::SendData(CRtMessageBlock& aData,
		BOOL aSendImmediately,
		RtRudpReliability aReliability,
		IRtTransport::RudpChannel aChannelID,
		INT aTTL)
{
	if (!m_bCanSend)
	{ return RT_ERROR_WRONG_STATUS; }

#if defined(RT_SUPPORT_BANDWIDTH_SERVER) || defined(RT_SUPPORT_BANDWIDTH)
	DWORD dwLen = aData.GetChainedLength();
#endif

	RtResult ret = m_transport->SendData(
		aData, aReliability, aTTL, aSendImmediately, 
		FALSE, IRtTransport::DP_HIGH, aChannelID);
#ifdef RT_SUPPORT_BANDWIDTH
	if(s_dwLastSendGetTick && RT_SUCCEEDED(ret))
	{
		s_dwSendData += dwLen;
	}
#endif

#if defined(RT_SUPPORT_BANDWIDTH_SERVER) || defined(RT_SUPPORT_BANDWIDTH)
	if(RT_SUCCEEDED(ret))
	{	m_statistic.SendData(dwLen); }
#endif	
	return ret;
}

RtResult CRtIMConnection::Disconnect(RtResult Reason)
{
	m_pSink = NULL;
	m_bCanSend = FALSE;
	RtResult ret = RT_OK;
	if (m_transport)
	{
		ret = m_transport->Disconnect(Reason);
		m_transport = NULL;
	}
	return ret;
}

RtResult	CRtIMConnection::GetOption(DWORD dwOption,
								LPVOID lpData)
{
#ifdef RT_SUPPORT_BANDWIDTH
	static DWORD 			m_dwCount = 0;
#endif	
	RT_ASSERTE_RETURN(lpData, RT_ERROR_INVALID_ARG);
// 	if (!m_transport)
// 	{ return RT_ERROR_FAILURE; }

	switch(dwOption)
	{
	case RT_OPT_TRANSPORT_PEER_ADDR:
		{
			TransportAddress* pAddr = (TransportAddress*)lpData;
			*pAddr = m_addrPeer;
			return m_addrPeer.strIP.empty() ? RT_ERROR_FAILURE : RT_OK;
		}
	case RT_OPT_TRANSPORT_LOCAL_ADDR:
		{
			TransportAddress* pAddr = (TransportAddress*)lpData;
			*pAddr = m_addrLocal;
			return m_addrLocal.strIP.empty() ? RT_ERROR_FAILURE : RT_OK;
		}
#ifdef RT_SUPPORT_BANDWIDTH
#define MIN_GETSPAN		1000
#define MAX_GETSPAN		15000
	case RT_CONF_OPT_SEND_BANDWITH:
		{
			m_dwCount++;
			if ( m_dwCount >150)
			{
				CRtString strMonitorString;
				CStatisticsData::GenerateMonitorInfo(strMonitorString);
				RT_INFO_TRACE(strMonitorString);
				m_dwCount = 0;
			}
			
			DWORD dwCount = (GetTickCountEx() - s_dwLastSendGetTick);
			if(dwCount >MIN_GETSPAN)
			{								
				LONGLONG lTmp = (LONGLONG)s_dwSendData * 8 * 1000;
				DWORD lastSendBand = lTmp /dwCount;
				s_dwSendDatas.push_back(lastSendBand);
				if (s_dwSendDatas.size() >MAX_DATA)
				{
					DWORD firstSendBand = s_dwSendDatas.front();
					s_dwSendDatas.pop_front();
					INT curSendBand = s_curSendBand * s_dwSendDatas.size() + lastSendBand - firstSendBand;
					if (curSendBand >0)
					{ s_curSendBand = curSendBand/s_dwSendDatas.size(); }
				}
				else
				{
					s_curSendBand = (s_curSendBand * (s_dwSendDatas.size()-1) + lastSendBand)/s_dwSendDatas.size();
				}
				if (dwCount >MAX_GETSPAN)
				{
					RT_WARNING_TRACE("CRtIMConnection::GetOption, too long after previous get send bandwith. time="<<dwCount);
				}
				s_dwLastSendGetTick = GetTickCountEx();
				s_dwSendData = 0;
			}
			// 				RT_WARNING_TRACE("CRtIMConnection::GetOption, sendv="<<s_curSendBand);			
			*((DWORD*)lpData) = s_curSendBand;
			return RT_OK;
		}
	case RT_CONF_OPT_RECV_BANDWITH:
		{
			DWORD dwCount = (GetTickCountEx() - s_dwLastRecvGetTick);
			if(dwCount >MIN_GETSPAN)
			{								
				LONGLONG lTmp = (LONGLONG)s_dwRecvData * 8 * 1000;
				DWORD lastRecvBand = lTmp /dwCount;
				s_dwRecvDatas.push_back(lastRecvBand);
				if (s_dwRecvDatas.size() >MAX_DATA)
				{
					DWORD firstRecvBand = s_dwRecvDatas.front();
					s_dwRecvDatas.pop_front();
					INT curRecvBand = s_curRecvBand * s_dwRecvDatas.size() + lastRecvBand - firstRecvBand;
					if (curRecvBand >0)
					{ s_curRecvBand = curRecvBand/s_dwRecvDatas.size(); }
				}
				else
				{
					s_curRecvBand = (s_curRecvBand * (s_dwRecvDatas.size()-1) + lastRecvBand)/s_dwRecvDatas.size();
				}
				if (dwCount >MAX_GETSPAN)
				{
					RT_WARNING_TRACE("CRtIMConnection::GetOption, too long after previous get recv bandwith. time="<<dwCount);
				}
				s_dwLastRecvGetTick = GetTickCountEx();
				s_dwRecvData = 0;
			}
			// 				RT_WARNING_TRACE("CRtIMConnection::GetOption, recev="<<s_curRecvBand);
			*((DWORD*)lpData) = s_curRecvBand;
			return RT_OK;
		}
#endif//!RT_SUPPORT_BANDWIDTH
	default:
		{
			if (!m_transport){
				return RT_ERROR_FAILURE; 
			}else{				
				return m_transport->GetOption(dwOption, lpData);
			}
		}
	}
	return RT_ERROR_NOT_IMPLEMENTED;
}
RtResult CRtIMConnection::SetOption(DWORD dwOption, LPVOID lpData)
{
	if (m_transport)
	{ m_transport->SetOption(dwOption, lpData);	}
	return RT_OK;
}

RtResult CRtIMConnection::GetAddress(DWORD dwOption, TransportAddress& transpAddress)
{
	CRtInetAddr addr;
	RtResult ret = m_transport->GetOption(dwOption,&addr);
	if(RT_FAILED(ret))
		return ret;

	transpAddress.strIP = addr.GetIpDisplayName();
	transpAddress.nPort = addr.GetPort();
	DWORD dwType;
	RT_ASSERTE_RETURN(m_transport.Get(), ret);
	ret = m_transport->GetOption(RT_OPT_TRANSPORT_TRAN_TYPE,&dwType);
	RT_ASSERTE(RT_SUCCEEDED(ret));
	if(dwType & CRtConnectionManager::CTYPE_TCP)
		transpAddress.nType = NET_PROTOCOL_TCP;
	else if(CRtConnectionManager::CTYPE_RUDP == dwType)
		transpAddress.nType = NET_PROTOCOL_RUDP;
	else if(dwType & CRtConnectionManager::CTYPE_UDP)
		transpAddress.nType = NET_PROTOCOL_UDP;
	else if(dwType & CRtConnectionManager::CTYPE_HTTP)
		transpAddress.nType = NET_PROTOCOL_HTTP;
	else
		transpAddress.nType = NET_PROTOCOL_NONE;
	return ret;
}

